package com.niit.service

import com.niit.utils.SparkUtils
import org.apache.spark.sql.{DataFrame, Dataset, Row, SaveMode}

/*
 对电影数据的具体分析过程

  需求：对电影评分数据进行分析，获得电影平均分Top10,要求电影次数大于200
 */
class MovieDataService {


  def dataAnalysis(): Unit ={
    //1.加载 数据文件 spark == SparkSession
    val spark = SparkUtils.takeSpark()
    val ds: Dataset[String] = spark.read.textFile("input/rating_100k.data")  //隐式转换
    import spark.implicits._ //隐式转换
    /*2.处理数据-- 获得电影id 和评分
      user  movie  rating       time
      196	  242	   3	       881250949

      将获得movieId 和 评分 变成一个数据表，为了后续可以使用SQL进行操作
         movieId  score
          242       3
          121       4
          242       2
     */
    val movieDF: DataFrame = ds.map(line => { // line:196	242	3	881250949
      val arrs: Array[String] = line.split("\t")
      val movieId = arrs(1) //获得每一个的电影id
      val score = arrs(2).toInt //获得每一个电影的评分
      //返回数据  元组 (电影id,评分)
      (movieId, score)
    }).toDF("movieId", "score")


    //3 根据需求  方式一： 编写SQL 进行数据分析  需求获得电影平均评分Top10 要求电影的次数大于200
    /*
          movieId   avgScore   counts
            194       4.4        232
     */
    movieDF.createOrReplaceTempView("movie")
    val sql =
        """
          | select movieId,avg(score) avgScore,count(*) counts
          |  from movie
          |  group by movieId
          |  having counts > 200
          |  order by avgScore desc
          |  limit 10
          |""".stripMargin
    val resDF: DataFrame = spark.sql(sql)

    //DF.write 保存数据库
    //应该写在dao层
    resDF.write
      .format( "jdbc") //连接MySql JDBC
      .option("url", "jdbc:mysql://node1:3306/BD2") //连接地址 MySQL数据库的版本 8.0+  localhost == 127.0.0.1== 本机IP
      .option("driver", "com.mysql.jdbc.Driver")//数据库驱动  8.0的驱动  com.mysql.cj.jdbc.Driver 5.0：com.mysql.jdbc.Driver
      .option("user", "root")//数据库的用户名
      .option("password", "Niit@123")//数据库的密码
      .option("dbtable", "movie")//数据库表
      .mode(SaveMode.Append)//当该表不存在的时候 会自动创建新表
      .save()//加载，将该表里面的数据封装到DataFrame中

    //3 根据需求  方式二： 编写DSL 进行数据分析   需求获得电影平均评分Top10 要求电影的次数大于200
    import org.apache.spark.sql.functions._  // 导入DSL的全部函数
    val resDF2: Dataset[Row] = movieDF.groupBy("movieId")
      .agg( //agg是对分组后的数据 进行统计和计算   agg一般都是和groupBy配合使用
        count("movieId") as ("counts"),
        avg('score) as ("avgScore")
      ).filter('counts > 200)
      .orderBy('avgScore.desc)
      .limit(10)
    //保存数据库
    resDF2.write
      .format( "jdbc") //连接MySql JDBC
      .option("url", "jdbc:mysql://node1:3306/BD2") //连接地址 MySQL数据库的版本 8.0+  localhost == 127.0.0.1== 本机IP
      .option("driver", "com.mysql.jdbc.Driver")//数据库驱动  8.0的驱动  com.mysql.cj.jdbc.Driver 5.0：com.mysql.jdbc.Driver
      .option("user", "root")//数据库的用户名
      .option("password", "Niit@123")//数据库的密码
      .option("dbtable", "movie")//数据库表
      .mode(SaveMode.Append)//当该表不存在的时候 会自动创建新表
      .save()//加载，将该表里面的数据封装到DataFrame中
  }

}
